Pulsar SQL 在智联搜索日志中的应用
🎙️ 阅读本文需 6 分钟
作者介绍：
高冉，智联招聘搜索团队研发工程师
>>> 业务背景 <<<
智联招聘作为国内知名的招聘平台之一，拥有 1.4 亿职场人用户，累计合作企业数达 400 万家。在如此繁忙的业务量下，公司对数据的搜索处理则变得更追求效率化。而在目前业务中，时常会遇到以下场景：
日常搜索服务的订阅问题中，需要根据注册 id 来获取现场的数据。
根据 C 端用户的投递信息，对职位投递的曝光途径进行一些分析。
B 端用户发布了很多职位，通过搜索匹配来为他们提供职位曝光，通过曝光的数据，检验 B 端用户的投入是否到位。
对搜索数据进行多维度聚合，生成更直观的图表，作为分析依据。
>>> 消息处理流程 <<<
智联系统中的消息处理流程如下所示：
首先将采集到的日志消息，如简历曝光、职位曝光等，进行统一的拆分。将这些请求与结果数据进行分离，保存在不同的 topic 中。
接下来对这些数据按城市/行业等维度进行聚合。处理完的数据会发送给 Pulsar，通过 Pulsar SQL 进行查询，图中间部分展示的是生成的 Pulsar 信息。上图粉色标注部分则主要会将中间的结果做成数据源，进行流量回放。
通过此消息处理系统，可以生成专门的职位曝光请求，提供给测试使用，用来模拟一些线上流量。还可以针对某一渠道的请求进行流量回放。或是对请求的参数，按测试要求进行修改，进行更有针对性的测试。
同时也可以对算法团队提供的模型进行反复测试，将不同模型的得分情况记录到 topic 中，使用 Pulsar SQL 查询，再提供给算法团队进行使用。
>>> Pulsar SQL 使用场景 <<<
>> 职位曝光查询
对一个职位在某一个时间段内的曝光情况进行查询。从上图可以看到每次曝光的打分情况，绿色部分则为曝光的产品参数。
>> 图表分析
通过消息系统对数据聚合后，可以制作一些图表（如热门城市），在时间段/行业等类别内进行横向对比。
>>> 为什么使用 Pulsar SQL <<<
如需在以上使用场景中生成类似的数据，其他产品也可以做到，为什么智联最后选择了 Pulsar SQL 呢？
智联是国内最早使用并落地 Pulsar 的公司，有成熟的 Pulsar 消息系统，可以将日志消息十分方便地与 Pulsar 结合。
Pulsar SQL 的技术栈简单，只要发送消息，就可以通过 SQL 语法进行查询。
>>> SQL 使用流程 <<<
接下来将从「发送消息」和「查询数据端」为大家介绍一下 Pulsar SQL 的使用流程，同时补充一些关于提升查询效率方面的技巧。
1. 根据场景，定义 javabean
与使用传统数据库类型映射相似，Pulsar SQL 也支持以下类型：基础数据类型、基础数据包装类型、String、类型嵌套。
2. 使用 producer 发送消息
如需使用 Pulsar SQL 特性，则需使用 Pulsar Schema 的特性（如下图）。使用这个特性也很简单，用定义好的 JavaBean 即可。
修改批量发送参数的配置（如上图），Pulsar 消息默认最大发送量为 1000，最大发送消息延迟为 1 毫秒。由于业务本身对发送延迟并没有这么高的要求，所以适当提高了发送延迟，这样就可以提高批量发送的效果。
最后开启数据压缩（如上图）。提高发送效果、开启压缩后，消息体量缩小为调整之前的五分之一，这样就可以在有限的空间内保存更多的数据。
🙋♂️TIPS：设置消息存留参数
提到保存数据，一般会有两个参数：保留时间（time）和最大保留空间（size），根据机器性能设置合适参数即可。
使用方法：通过 pulsar admin set-retention 命令来对存留相关指标进行设置。通过 get-retention 命令来查看设置是否生效与合理。
pulsar-admin namespaces set-retention --size 1g --time 3d sql/testpulsar-admin namespaces get-retention sql/test需要注意一点，默认情况下 topic 如果没有消费者，Pulsar 则不会保留数据。如果设置了存留策略，则会按策略进行。
>> 查询数据
上图展示了一天内，通过日志消费加工出的所有日志总和。发送完消息后，接下来就可以进行查询。
1. 使用 Presto 命令查询
在 Pulsar 程序里，执行 Pulsar SQL 命令，进入命令行交互界面。
需要注意的是，类型嵌套属性名称比较特别，需要嵌套名称和类型名称进行组合（如上图）。除自定义的字段外，还有 Pulsar 消息的原数据字段。比如 partition（partition 消息）、publish_time （消息发送时间）等。
2. 使用 Presto 的 HTTP 接口查询
接口地址：http://ip:port/v1/statement
注意，请求时我们还需要加上头信息 X-Presto-User，同时在请求体中还要加入 SQL。
得到的返回数据结构如下所示：
id 为查询 id，出现 nextUri 说明此次查询还未完成，可以循环调用这个接口，直到 nextUri 消失。coluumns 字段为返回表头的信息。data 是返回的数据，对此需要进行循环的收集，stats 是 presto 接口的状态。
下图展示了用 Presto 接口做的查询效果，上部分显示的是一些状态信息：扫描了多少数据、用了多长时间等，这些请求数据都会进行实时变化。
此图因涉及公司隐私信息，遂打码处理
3. 使用 Metabase 查询
Metabase 是一个开源的数据探索工具，支持多种数据库/源，生成各种形式的图表。使用方法：设置好数据源，选择数据库类型 Presto，连接 IP，设置端口（8081）。设置完成后即可用 Metabase 进行查询。
下图为使用 Metabase 生成的一些图表：
>> 提升查询效率
现实情况中，如果查询消息量很大，从中找到需要的数据是很困难的，这需要扫描很多的数据，耗费大量的时间，效率极低。
1. 改进方法：增加 partition 数量
此图因涉及公司隐私信息，遂打码处理
可以采用分区策略。例如将职位数据按职位 ID 进行分区，这样可以成倍提升查询效果。
select traceid, _partition_ from "sql|stat"."position-monitor-detail-jdid-partion-1" limit 100select traceid, _partition_ from "sql|stat"."position-monitor-detail-jdid" where _partition_ = 1 limit 100select traceid, _partition_ from "sql|stat"."position-monitor-detail-jdid" where _partition_ in (1,2) limit 100刚开始使用 Pulsar SQL 时，只能按真实 topic 名称来查询。但其实在 Pulsar 里，原始的 topic 名称不是真实的 topic 名称。在经过社区的改进后，增加了源属性。这样就可以将 partition 当作搜索字段，放入搜索条件进行查询。
2. 改进方法：限定 publishTime 范围
publish_time 是源数据的一个字段，可以将它放置在 “where” 条件后。
select * from "sql|stat"."position-monitor-detail-jdid"where _partition_ = 1and _publish_time_ > timestamp '2019-11-15 09:00:00'limit 100加上 publish_time 范围条件后，Pulsar SQL 可以在扫描数据时使用二分法来获取 ledger 的起始点，从而减少数据的扫描范围，大幅提升查询效率。
>>> Pulsar SQL 使用总结 <<<
目前 Pulsar SQL 的使用场景，均为一些问题排查的数据查找和聚合分析。同时帮助招聘业务两端用户使用场景的搜索变得更加高效。但由于目前 Pulsar SQL 缺少二级索引，所以灵活性有待提高。
>>> 展望 <<<
使用 Pulsar SQL 来进行消息的查询，比如指定 MessageId、未消费消息。
与 Pulsar Function 结合使用，更高效地进行数据处理、聚合。
增加更多的使用场景，比如查看热门职位、热门简历、热门搜索词等给 C 端、B 端用户带来更好的体验。
>>> 总结 <<<
通过使用 Pulsar SQL，智联招聘将原本大体量的搜索业务进行了效率化提升，解决了日常搜索场景中的部分问题，如保存更多的数据、查询数据过程中更高效等，均取得了明显的效果。
也可以看到，在运行强搜索的使用场景/行业中，因其技术栈简单而使 Pulsar SQL 成为一个更具性价比的选择。